package com.doit.day02;

import org.apache.commons.lang3.RandomUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaProducerDemo {
    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        /**
         * 必须配置的
         */
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux01:9092,linux02:9092,linux03:9092");
        props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        /**
         * 选配
         */
//        partitioner.class   分区器
        props.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG,MyPartitioner.class.getName());
//        acks
        props.setProperty(ProducerConfig.ACKS_CONFIG,"-1");

        //创建一个生产者对象==》 conf
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
        for (int i = 0; i < 1000; i++) {
            //准备发送的数据
            ProducerRecord<String, String> record = new ProducerRecord<>("java", "this message from idea ==> "+i);
            //调用api发送数据
            producer.send(record);
            //刷新一下
            producer.flush();
            Thread.sleep(RandomUtils.nextInt(100,200));
        }
        //关流
        producer.close();

    }
}
